#include <new>
#include <stdio.h>
#include "process_sharefile.h"

#define SHARE_FILE_HEAD_MAGIC (0x055667788)
#define SHARE_FILE_DATA_HEAD_MAGIC (0xaabbccdd)

#define CONNECTSTATUSCOR 1 //连接成功
#define CONNECTSTATUSERR 0 //连接失败
#define READ_ERRORMAXTIMES 5
#define SELECT_TIMEOUT 20

typedef struct
{
    u32 magic;   //魔幻数
    u32 datalen; //事件数据长度
    // u8  data[0]; 		//数据
} TProcessShareDataHead;

CProcessShare::CProcessShare()
{
    m_hFile = TZTEK_MEMFD_INVALID_FILE;
    m_pBaseAddr = NULL;
    m_pProcessShareCtx = NULL;
    m_fd = -1;
    m_dwlistenFd = -1;
    m_connectStatus = CONNECTSTATUSERR;
    m_bflag = false;
    m_dwErrorCnt = 0;
    m_dwSelectTimeout = 0;
}

CProcessShare::~CProcessShare()
{
}

bool TZTEK_Truncate(FILEHANDLE hFile, u32 dwSize)
{
    return ftruncate(hFile, dwSize) == 0 ? true : false;
}
bool TZTEK_UnMapFile(void* pAddr, u32 dwLen)
{
    return (munmap(pAddr, dwLen) == 0) ? true : false;
}
bool TZTEK_DeleteFile(const char* pFileName)
{
    return remove(pFileName) == 0 ? true : false;
}
bool TZTEK_CloseFile(FILEHANDLE hFile)
{
    return close(hFile) == 0 ? true : false;
}
u32  TZTEK_GetLastError()
{
    return errno;
}

void* TZTEK_MapFile(FILEHANDLE hFile, u32 dwProt, u32 nFlag, u32 dwSize, const char *pMapName)
{
    void* pRet = mmap(NULL, dwSize, dwProt, nFlag, hFile, 0);
    if (pRet == MAP_FAILED)
    {
        return NULL;
    }
    return pRet;
}

bool CProcessShare::InitClient()//这里是客户端
{
    if (InitClientSocket(m_strFileName.c_str()) == false)
    {
        return false;
    }

    m_connectStatus = CONNECTSTATUSCOR;
    int hFile = -1;
    hFile = memfd_create(m_strFileName.c_str(), MFD_ALLOW_SEALING);
    if (hFile == TZTEK_MEMFD_INVALID_FILE)
    {
        printf("memfd_create  failed,filename=%s,sharename=%s,errno = %u\n", m_strFileName.c_str(), m_strShareFileName.c_str(), TZTEK_GetLastError());
        return false;
    }
    m_hFile = hFile;

    send_fd(m_fd, m_hFile, 1);

    void *pBaseAddr = TZTEK_MapFile(m_hFile, TZTEK_PROT_READWRITE, TZTEK_MAP_SHARED, m_dwFileSize, m_strShareFileName.c_str());
    if (pBaseAddr == NULL)
    {
        TZTEK_CloseFile(m_hFile);
        printf("TZTEK_MapFile failed,filename=%s,m_hFile=%d,errno = %u\n", m_strFileName.c_str(), m_hFile, TZTEK_GetLastError());
        return false;
    }

    m_pBaseAddr = (u8 *)pBaseAddr;
    m_pProcessShareCtx = (TProcessShareHead *)pBaseAddr;
    m_pDataAddr = m_pBaseAddr + sizeof(TProcessShareHead);

    /*  不论订阅还是发布，由于在不同进程，均需要初始化数据  */
    TZTEK_Truncate(m_hFile, m_dwFileSize);
    m_pProcessShareCtx->dwMagic = SHARE_FILE_HEAD_MAGIC;
    m_pProcessShareCtx->dwSumSize = m_dwFileSize;
    m_pProcessShareCtx->dwTailLen = m_dwMaxDataPacket;
    m_pProcessShareCtx->dwRingLen = m_dwFileSize - sizeof(TProcessShareHead) - m_dwMaxDataPacket;
    m_pProcessShareCtx->dwRidx = 0;
    m_pProcessShareCtx->dwWidx = 0;
    m_bflag = true;
    printf("*********topic : %s connection succeeded************\n", m_strFileName.c_str());

    return true;
}

void *CProcessShare::createPubSocketThread()
{
    while (1)
    {
        if (InitClient())
        {
            return 0;
        }
        sleep(10);
    }
}

void *CProcessShare::createSubSocketThread()
{
    while (1)
    {
        //此处连接完之后还要处于监听状态，以免客户端退出了再次
        InitServer();

        sleep(10);
    }
}

bool CProcessShare::InitServer() //这里是服务端
{
    /* 添加重连机制 */
    InitServerSocket(m_strFileName.c_str());

    m_connectStatus = CONNECTSTATUSCOR;
    printf("*********topic : %s  connection succeeded************\n", m_strFileName.c_str());
    void *pBaseAddr = TZTEK_MapFile(m_hFile, TZTEK_PROT_READWRITE, TZTEK_MAP_SHARED, m_dwFileSize, m_strShareFileName.c_str());
    if (pBaseAddr == NULL)
    {
        TZTEK_CloseFile(m_hFile);
        printf("TZTEK_MapFile failed,filename=%s,m_hFile=%d,errno = %u\n", m_strFileName.c_str(), m_hFile, TZTEK_GetLastError());
        return false;
    }

    m_pBaseAddr = (u8 *)pBaseAddr;
    m_pProcessShareCtx = (TProcessShareHead *)pBaseAddr;
    m_pDataAddr = m_pBaseAddr + sizeof(TProcessShareHead);

    /*  不论订阅还是发布，由于在不同进程，均需要初始化数据  */
    TZTEK_Truncate(m_hFile, m_dwFileSize);
    m_pProcessShareCtx->dwMagic = SHARE_FILE_HEAD_MAGIC;
    m_pProcessShareCtx->dwSumSize = m_dwFileSize;
    m_pProcessShareCtx->dwTailLen = m_dwMaxDataPacket;
    m_pProcessShareCtx->dwRingLen = m_dwFileSize - sizeof(TProcessShareHead) - m_dwMaxDataPacket;
    m_pProcessShareCtx->dwRidx = 0;
    m_pProcessShareCtx->dwWidx = 0;

    m_bflag = true;
    

    return true;
}

bool CProcessShare::InitServerSocket(const s8 *pFileName)
{
    do
    {
        if (m_dwlistenFd <= 0)
        {
            m_dwlistenFd = getSocket(pFileName);
            if (m_dwlistenFd <= 0)
            {
                printf("Init socket [%s] failed with error %d\n", pFileName, errno);
                continue;
            }
        }
        m_fd = accept(m_dwlistenFd, (struct sockaddr *)NULL, NULL);
        if (m_fd < 0)
        {
            printf("accept to [%s] failed with error %d\n", pFileName, errno);
            continue;
        }

        m_hFile = recv_fd(m_fd);

    } while (m_hFile == 0);

    return true;
}

/*pShareName 是为了适配windows的，linux是不需要的*/
bool CProcessShare::Init(const s8 *pFileName, u32 dwFileSize, const s8 *pShareName, u32 dwMaxDataPacket, bool bProducer)
{
    if (pFileName == NULL || pShareName == NULL)
    {
        printf("filename is invalid\n");
        return false;
    }
    if (dwFileSize == 0)
    {
        printf("filesize is invalid\n");
        return false;
    }
    /* Create an anonymous file in tmpfs  ,  allow seals to be placed on the file. */
    m_strFileName = pFileName;
    m_dwFileSize = dwFileSize;
    m_strShareFileName = pShareName;
    m_dwMaxDataPacket = dwMaxDataPacket;

    if (bProducer)
    {
        m_pPubSocketThread = new std::thread(&CProcessShare::createPubSocketThread, this);
        m_pPubSocketThread->detach();
    }
    else
    {
        m_pSubSocketThread = new std::thread(&CProcessShare::createSubSocketThread, this);
        m_pSubSocketThread->detach();
    }
    return true;
}

bool CProcessShare::InputData(u8 *pData, u32 dwDataLen)
{
    if (m_bflag != true)
    {
        // printf("topic : %s  connect failed  , can not InputData  !\n ", m_strFileName.c_str());
        return false;
    }
    if (m_connectStatus != CONNECTSTATUSCOR)
    {
        return false;
    }
    u32 dwWriteLen = 0;
    u8 *pDest = NULL;
    TProcessShareDataHead head;

    if (dwDataLen == 0)
    {
        printf("invalid input para,dwDataLen = %u\n", dwDataLen);
        return false;
    }

    /*校验缓存空间*/
    dwWriteLen = sizeof(TProcessShareDataHead) + dwDataLen;
    if (checkWritable(dwWriteLen) == false)
    {
        printf("topic : %s there is no space,dwDataLen = %u\n", m_strFileName.c_str(), dwDataLen);
        m_dwErrorCnt++;
        if (m_dwErrorCnt > READ_ERRORMAXTIMES)
        {
            close(m_fd); //没数据时，可能是断线了，需要关闭socket
            m_connectStatus = CONNECTSTATUSERR;
            m_dwErrorCnt = 0;
        }
        return false;
    }

    /*写入缓存*/
    memset(&head, 0, sizeof(TProcessShareDataHead));
    head.magic = SHARE_FILE_DATA_HEAD_MAGIC;
    head.datalen = dwDataLen;
    pDest = m_pDataAddr + m_pProcessShareCtx->dwWidx;
    memcpy(pDest, &head, sizeof(TProcessShareDataHead));
    memcpy(pDest + sizeof(TProcessShareDataHead), pData, dwDataLen);
    /*更新写指针*/
    if (m_pProcessShareCtx->dwWidx + dwDataLen >= m_pProcessShareCtx->dwRingLen)
    {
        m_pProcessShareCtx->dwWidx = 0;
    }
    else
    {
        m_pProcessShareCtx->dwWidx += dwWriteLen;
    }

    return true;
}

bool CProcessShare::InputData(u8 *pData1, u32 dwDataLen1, u8 *pData2, u32 dwDataLen2)
{
    if (m_bflag != true)
    {
        printf("topic : %s  connect failed  , can not InputData  !\n ", m_strFileName.c_str());
        return false;
    }

    if (m_connectStatus != CONNECTSTATUSCOR)
    {
        return false;
    }

    u32 dwWriteLen = 0;
    u8 *pDest = NULL;
    TProcessShareDataHead head;

    if (dwDataLen1 == 0 && dwDataLen2 == 0)
    {
        printf("invalid input para\n");
        return false;
    }

    /*校验缓存空间*/
    dwWriteLen = sizeof(TProcessShareDataHead) + dwDataLen1 + dwDataLen2;
    if (checkWritable(dwWriteLen) == false)
    {
        printf("topic : %s there is no space,dwDataLen = %u m_dwErrorCnt is %d \n", m_strFileName.c_str(), dwDataLen1 + dwDataLen2,m_dwErrorCnt);
        m_dwErrorCnt++;
        if (m_dwErrorCnt > READ_ERRORMAXTIMES)
        {
            close(m_fd); //没数据时，可能是断线了，需要关闭socket
            m_connectStatus = CONNECTSTATUSERR;
            m_dwErrorCnt = 0;
        }

        return false;
    }

    /*写入缓存*/
    memset(&head, 0, sizeof(TProcessShareDataHead));
    head.magic = SHARE_FILE_DATA_HEAD_MAGIC;
    head.datalen = dwDataLen1 + dwDataLen2;
    pDest = m_pDataAddr + m_pProcessShareCtx->dwWidx;
    memcpy(pDest, &head, sizeof(TProcessShareDataHead));
    if (pData1)
    {
        memcpy(pDest + sizeof(TProcessShareDataHead), pData1, dwDataLen1);
    }
    if (pData2)
    {
        memcpy(pDest + sizeof(TProcessShareDataHead) + dwDataLen1, pData2, dwDataLen2);
    }

    /*更新写指针*/
    if (m_pProcessShareCtx->dwWidx + dwWriteLen >= m_pProcessShareCtx->dwRingLen)
    {
        m_pProcessShareCtx->dwWidx = 0;
    }
    else
    {
        m_pProcessShareCtx->dwWidx += dwWriteLen;
    }

    // printf("%s:m_dwRidx = %d,m_dwWidx=%d,dwWiterlen is %d \n\n", __func__, m_pProcessShareCtx->dwRidx, m_pProcessShareCtx->dwWidx, dwWriteLen);
    return true;
}

bool CProcessShare::GetBuf(u8 *&pBuf, u32 &dwBufLen)
{
    u32 dwSpace = m_pProcessShareCtx->dwRingLen - (m_pProcessShareCtx->dwRidx - m_pProcessShareCtx->dwRidx + m_pProcessShareCtx->dwRingLen) % m_pProcessShareCtx->dwRingLen;

    if (m_pProcessShareCtx->dwWidx > m_pProcessShareCtx->dwRidx && m_pProcessShareCtx->dwRidx > 0)
    {
        dwSpace += m_pProcessShareCtx->dwTailLen;
    }
    if (dwSpace <= sizeof(TProcessShareDataHead))
    {
        dwBufLen = 0;
        return false;
    }
    //头由内部自动填充
    dwBufLen = dwSpace - sizeof(TProcessShareDataHead) - 1;
    pBuf = m_pDataAddr + m_pProcessShareCtx->dwWidx + sizeof(TProcessShareDataHead);
    printf("%s:m_dwRidx = %d,m_dwWidx=%d,dwSpace=%d\n", __func__, m_pProcessShareCtx->dwRidx, m_pProcessShareCtx->dwWidx, dwSpace);
    return true;
}

bool CProcessShare::GetData(u8 *&pData, u32 &dwDatalen)
{
    if (m_bflag != true)
    {
        return false;
    }
    bool bRet = false;
    TProcessShareDataHead *pHead = NULL;
    u32 dwLen = 0;
    dwLen = (m_pProcessShareCtx->dwWidx - m_pProcessShareCtx->dwRidx + m_pProcessShareCtx->dwRingLen) % m_pProcessShareCtx->dwRingLen;
    do
    {
        /*有数据直接返回*/
        if (dwLen >= sizeof(TProcessShareDataHead))
        {
            bRet = true;
            m_dwErrorCnt = 0;
            pHead = (TProcessShareDataHead *)(m_pDataAddr + m_pProcessShareCtx->dwRidx);
            break;
        }

        /*校验拖尾数据是否可读*/
        if (m_pProcessShareCtx->dwRidx > m_pProcessShareCtx->dwWidx)
        {
            bRet = true;
            m_dwErrorCnt = 0;
            pHead = (TProcessShareDataHead *)(m_pDataAddr + m_pProcessShareCtx->dwRidx);
            break;
        }
        bRet = false;
        m_dwErrorCnt++;
        if (m_dwErrorCnt > 30*1000*1000)//每1ms获取一次数据，因此这里的次数需要大一点
        {
            close(m_fd); //没数据时，可能是断线了，需要关闭socket
            m_connectStatus = CONNECTSTATUSERR;
            m_dwErrorCnt = 0;
        }
        

    } while (0);

    if (bRet)
    {
        pData = (u8 *)(pHead + 1);
        dwDatalen = pHead->datalen;
        printf("topic : %s , %s:m_dwRidx = %d,m_dwWidx=%d,dwDatalen=%d\n", m_strFileName.c_str(), __func__, m_pProcessShareCtx->dwRidx, m_pProcessShareCtx->dwWidx, dwDatalen);
    }
    return bRet;
}

bool CProcessShare::UpdateReadIdx(u32 dwLen)
{
    u32 dwReadLen = dwLen + sizeof(TProcessShareDataHead);
    if (m_pProcessShareCtx->dwRidx + dwReadLen >= m_pProcessShareCtx->dwRingLen)
    {
        m_pProcessShareCtx->dwRidx = 0;
    }
    else
    {
        m_pProcessShareCtx->dwRidx += dwReadLen;
    }
    return true;
}

bool CProcessShare::UpdateWriteIdx(u32 dwLen)
{
    TProcessShareDataHead *head = (TProcessShareDataHead *)(m_pDataAddr + m_pProcessShareCtx->dwWidx);
    head->magic = SHARE_FILE_DATA_HEAD_MAGIC;
    head->datalen = dwLen;

    u32 dwRealLen = dwLen + sizeof(TProcessShareDataHead);
    if (m_pProcessShareCtx->dwWidx + dwRealLen >= m_pProcessShareCtx->dwRingLen)
    {
        m_pProcessShareCtx->dwWidx = 0;
    }
    else
    {
        m_pProcessShareCtx->dwWidx += dwRealLen;
    }
    printf("%s:m_dwRidx = %d,m_dwWidx=%d,m_dwRingLen=%d,dwRealLen=%d\n", __func__, m_pProcessShareCtx->dwRidx, m_pProcessShareCtx->dwWidx, m_pProcessShareCtx->dwRingLen, dwRealLen);
    return true;
}

bool CProcessShare::checkWritable(u32 dwLen)
{
    u32 dwSpace = m_pProcessShareCtx->dwRingLen - (m_pProcessShareCtx->dwWidx - m_pProcessShareCtx->dwRidx + m_pProcessShareCtx->dwRingLen) % m_pProcessShareCtx->dwRingLen;
    if (m_pProcessShareCtx->dwWidx > m_pProcessShareCtx->dwRidx && m_pProcessShareCtx->dwRidx > 0)
    {
        dwSpace += m_pProcessShareCtx->dwTailLen;
    }
    return (dwSpace > dwLen);
}

bool CProcessShare::Release()
{
    if (m_pBaseAddr)
    {
        TZTEK_UnMapFile(m_pBaseAddr, 2 * 1024 * 1024);
    }

    if (m_hFile != TZTEK_MEMFD_INVALID_FILE)
    {
        TZTEK_CloseFile(m_hFile);
        TZTEK_DeleteFile(m_strFileName.c_str());
    }

    m_hFile = TZTEK_MEMFD_INVALID_FILE;
    m_pProcessShareCtx = NULL;
    m_pBaseAddr = NULL;
    m_pDataAddr = NULL;

    if (m_fd != -1)
    {
        close(m_fd);
        m_fd = -1;
    }

    return true;
}

bool CProcessShare::InitClientSocket(const s8 *pFileName)
{
    if (-1 != m_fd)
    {
        close(m_fd);
        m_fd = -1;
    }

    m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (m_fd < 0)
    {
        printf("Failed to create socket\n");
        return false;
    }

    struct sockaddr_un addr; //指定路径名
    memset(&addr, 0, sizeof(struct sockaddr_un));
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, pFileName, sizeof(addr.sun_path) - 1);

    /* 客户端需要用connect()建立连接 */
    if (connectWithTimeOut(m_fd, (struct sockaddr *)&addr, 300) < 0)
    {
        printf("topic : %s failed to connect to socket, with error %d socket error %s\n", pFileName, errno, strerror(errno));
        return false;
    }
    m_connectStatus = CONNECTSTATUSCOR;
    return true;
}

int CProcessShare::connectWithTimeOut(int fd, struct sockaddr *pAddr, unsigned int dwTimeOut)
{
    int n = -1, ret = -1, err = 0;
    socklen_t len = -1;

    int flag = fcntl(fd, F_GETFL, 0); //取得fd的文件状态标志
    fcntl(fd, F_SETFL, flag | O_NONBLOCK);

    n = connect(fd, (struct sockaddr *)pAddr, sizeof(struct sockaddr_un)); //客户端需要用connect()建立连接
    if (0 == n)
    {
        goto success;
    }
    else
    {
        if (errno != EINPROGRESS)
        {
            goto done;
        }
    }
    fd_set wset;       //是一组文件描述字(fd)的集合，它用一位来表示一个fd
    FD_ZERO(&wset);    //将指定的文件描述符集清空，在对文件描述符集合进行设置前，必须对其进行初始化
    FD_SET(fd, &wset); //用于在文件描述符集合中增加一个新的文件描述符
    struct timeval tv; //用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为0
    tv.tv_sec = dwTimeOut / 1000;
    tv.tv_usec = 1000 * (dwTimeOut % 1000);
    n = select(fd + 1, NULL, &wset, NULL, &tv); //检查套节字是否可读,循环询问文件节点，超时时间到了就跳过代码继续往下执行,返回对应位仍然为1的fd的总数
    if (n <= 0)
    {
        goto done;
    }
    if (FD_ISSET(fd, &wset)) //检查fd是否在这个集合里面,当检测到fd在set中则返回真，否则，返回假（0）
    {
        len = sizeof(err);
        n = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len); //获取一个套接字的选项，并将结果储存到&err中
        if (n < 0 || err)
        {
            goto done;
        }
    }
    else
    {
        goto done;
    }

success:
    ret = 0;
done:
    fcntl(fd, F_SETFL, flag); //设置文件的flag，fd依然处于非阻塞状态，方便下次重连
    return ret;
}

s32 CProcessShare::getSocket(const s8 *pFileName)
{
    struct sockaddr_un addr; //指定路径名
    memset(&addr, 0, sizeof(struct sockaddr_un));
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, pFileName, sizeof(addr.sun_path) - 1);

    int listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (listen_fd < 0)
    {
        printf("socket error %s errno: %d\n", strerror(errno), errno);
        return -1;
    }

    if (unlink(pFileName) == -1)
    {
        printf("Removing socket file :%s  failed: %s\n", pFileName, strerror(errno));
    }

    int ret = -1;
    ret = bind(listen_fd, (struct sockaddr *)&addr, sizeof(addr)); //这一步会创建文件,并调用
    if (ret < 0)
    {
        printf("bind socket error %s errno: %d\n", strerror(errno), errno);
        return ret;
    }

    ret = listen(listen_fd, 1024);
    if (ret < 0)
    {
        printf("listen error %s errno: %d\n", strerror(errno), errno);
        return ret;
    }
    return listen_fd;
}

void CProcessShare::send_fd(int socket, int fd, int n) // send fd by socket
{
    struct msghdr msg = {0};
    struct cmsghdr *cmsg;
    char buf[CMSG_SPACE(n * sizeof(int))], data;
    memset(buf, 0, sizeof(buf));
    struct iovec io = {.iov_base = &data, .iov_len = 1};

    msg.msg_iov = &io;     /*多io缓冲区的地址*/
    msg.msg_iovlen = 1;    /*缓冲区的个数*/
    msg.msg_control = buf; //指向与协议控制相关的消息或者辅助数据
    msg.msg_controllen = sizeof(buf);

    cmsg = CMSG_FIRSTHDR(&msg);                 /*返回msghdr辅助数据部分指向第一个cmsghdr的指针*/
    cmsg->cmsg_level = SOL_SOCKET;              /*具体的协议标识  */
    cmsg->cmsg_type = SCM_RIGHTS;               /*协议中的类型*/
    cmsg->cmsg_len = CMSG_LEN(n * sizeof(int)); /*包含该头部的数据长度*/

    memcpy((int *)CMSG_DATA(cmsg), &fd, n * sizeof(int)); // CMSG_DATA返回cmsghdr的净荷数据部分

    if (sendmsg(socket, &msg, 0) < 0)
    {
        printf("Failed to send message\n");
    }
}

int CProcessShare::recv_fd(int socket)
{
    int fd;
    struct msghdr msg = {0};
    struct cmsghdr *cmsg;

    char buf[CMSG_SPACE(sizeof(int))], data;
    memset(buf, '\0', sizeof(buf));
    struct iovec io = {.iov_base = &data, .iov_len = 1};

    msg.msg_iov = &io; /*多io缓冲区的地址*/
    msg.msg_iovlen = 1;
    msg.msg_control = buf;
    msg.msg_controllen = sizeof(buf);

    if (recvmsg(socket, &msg, 0) == 0)
    {
        printf("Failed to receive message  \n");
        return 0;
    }
    cmsg = CMSG_FIRSTHDR(&msg);
    fd = *(int *)CMSG_DATA(cmsg);

    return fd;
}
